package com.hub.realtime.common.utils;

import com.alibaba.fastjson.JSON;
import com.hub.realtime.common.core.domain.model.ClusterInfo;
import com.hub.realtime.common.enums.JobRunState;
import com.hub.realtime.common.model.FlinkRestModel;
import com.hub.realtime.common.model.JobInfo;
import com.hub.realtime.common.utils.hadoop.HadoopUtil;
import com.hub.realtime.common.utils.http.HttpUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.YarnApplicationState;

import java.util.ArrayList;
import java.util.List;

@Slf4j
public class JobUtils {
    public static FlinkRestModel getJobInfo(ClusterInfo clusterInfo, ApplicationId applicationId) {
        FlinkRestModel flinkRestModel = new FlinkRestModel();
        JobInfo jobInfo = new JobInfo();
        List<JobInfo> jobs = new ArrayList<>();
        YarnAdapter yarnAdapter = new YarnAdapter();
        YarnApplicationState applicationState = yarnAdapter.getApplicationState(clusterInfo.getHadoopConfDir(),applicationId);
        if (applicationState == null) {
            jobInfo.setState(JobRunState.FINISHED.name());
            jobInfo.setJid(" ");
            jobInfo.setStartTime(null);
            jobs.add(jobInfo);
            flinkRestModel.setJobs(jobs);
        } else if (applicationState == YarnApplicationState.FAILED
                || applicationState == YarnApplicationState.FINISHED || applicationState == YarnApplicationState.KILLED) {
            if (applicationState == YarnApplicationState.FAILED) {
                jobInfo.setState(JobRunState.FAILED.name());
            } else {
                jobInfo.setState(JobRunState.FINISHED.name());
            }
            jobInfo.setJid(" ");
            jobInfo.setStartTime(null);
            jobs.add(jobInfo);
            flinkRestModel.setJobs(jobs);
        } else {
            try {
                String resUrl =  HadoopUtil.getRMWebAppURL(clusterInfo,true).concat("/proxy/").concat(applicationId.toString()).concat("/jobs/overview");
                String res = HttpUtils.sendGet(resUrl);
                if (StringUtils.isNotEmpty(res)) {
                    flinkRestModel = JSON.parseObject(res, FlinkRestModel.class);
                }else
                {
                    jobs.add(jobInfo);
                    flinkRestModel.setJobs(jobs);
                }
            } catch (Exception ex) {
                jobs.add(jobInfo);
                flinkRestModel.setJobs(jobs);
                log.error("获取作业状态出错：" + ex.getMessage());
                ex.printStackTrace();
                return flinkRestModel;
            }
        }
        return flinkRestModel;
    }

}

